package drds.server.datasource;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.config.Alarms;
import drds.server.config.model.GroupDataSourceConfig;
import drds.server.config.model.MysqlInstanceInfo;
import drds.server.handler.ConnectionHeartBeatHandler;
import drds.server.handler.DelegateResponseHandler;
import drds.server.handler.Handler;
import drds.server.handler.NewConnectionRespHandler;
import drds.server.net.SessionContextGroupShareResources;
import drds.server.util.CurrentTimeMillis;

public class DataSource {
	static {
		try {
			Class.forName("com.mysql.jdbc.Driver");
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
			throw new RuntimeException("请将mysql驱动放置在classpath路径下", e);
		}
	}
	private static final Logger LOGGER = LoggerFactory.getLogger(DataSource.class);

	private final String id;
	private final MysqlInstanceInfo mysqlInstanceInfo;
	private final int maxSize;
	private final boolean isReadNode;
	private final SchemaToConnectionQueueMap schemaConnectionHolderQueueMap = new SchemaToConnectionQueueMap();
	private final GroupDataSourceConfig groupDataSourceConfig;
	private GroupDataSource groupDataSource;
	private AtomicLong writeCount = new AtomicLong(0);
	private AtomicLong readCount = new AtomicLong(0);

	/**
	 * 心跳线程作为后台线程自动运行。
	 */
	private volatile long heartbeatRecoveryTime;
	private final ConnectionHeartBeatHandler conHeartBeatHanler = new ConnectionHeartBeatHandler();

	public DataSource(MysqlInstanceInfo mysqlInstanceInfo, GroupDataSourceConfig groupDataSourceConfig, boolean isReadNode) {
		this.id = mysqlInstanceInfo.getId();
		this.mysqlInstanceInfo = mysqlInstanceInfo;
		this.maxSize = mysqlInstanceInfo.getMaxSize();
		this.groupDataSourceConfig = groupDataSourceConfig;
		this.isReadNode = isReadNode;
	}

	public long getReadCount() {
		return readCount.get();
	}

	public void addReadCount() {
		readCount.addAndGet(1);
	}

	public long getWriteCount() {
		return writeCount.get();
	}

	public void addWriteCount() {
		writeCount.addAndGet(1);
	}

	public GroupDataSourceConfig getGroupDataSourceConfig() {
		return groupDataSourceConfig;
	}

	public boolean isReadNode() {
		return isReadNode;
	}

	public int getMaxSize() {
		return maxSize;
	}

	public void setGroupDataSource(GroupDataSource groupDataSource) {
		this.groupDataSource = groupDataSource;
	}

	public GroupDataSource getGroupDataSource() {
		return groupDataSource;
	}

	public String getId() {
		return id;
	}

	public int getActiveCountForSchema(String schema) {
		return schemaConnectionHolderQueueMap.getActiveCountOfSchema(schema, this);
	}

	public int getIdleCountForSchema(String schema) {
		ConnectionHolderQueue connectionHolderQueue = schemaConnectionHolderQueueMap.getConnectionHolderQueue(schema);
		return connectionHolderQueue.getConnectionHolderQueue().size();
	}

	public int getIdleCount() {
		int totalSize = 0;
		for (ConnectionHolderQueue connectionHolderQueue : schemaConnectionHolderQueueMap.values()) {
			totalSize += connectionHolderQueue.getConnectionHolderQueue().size();
		}
		return totalSize;
	}

	private boolean validSchema(String schema) {
		String theSchema = schema;
		return theSchema != null && !"".equals(theSchema) && !"snyn...".equals(theSchema);
	}

	private void checkIfNeedHeartBeat(LinkedList<IConnectionHolder> connectionHolderList, ConnectionHolderQueue connectionHolderQueue, ConcurrentLinkedQueue<IConnectionHolder> checkConnectionHolderList, long hearBeatTime, long hearBeatTime2) {
		int maxConsInOneCheck = 10;
		Iterator<IConnectionHolder> iterator = checkConnectionHolderList.iterator();
		while (iterator.hasNext()) {
			IConnectionHolder connectionHolder = iterator.next();
			if (connectionHolder.isClosedOrQuit()) {
				iterator.remove();
				continue;
			}
			if (validSchema(connectionHolder.getSchema())) {
				if (connectionHolder.getLastTime() < hearBeatTime && connectionHolderList.size() < maxConsInOneCheck) {
					iterator.remove();
					// Heart beat check
					connectionHolder.setBorrowed(true);
					connectionHolderList.add(connectionHolder);
				}
			} else if (connectionHolder.getLastTime() < hearBeatTime2) {
				// not valid schema conntion should close for idle
				// exceed 2*conHeartBeatPeriod
				iterator.remove();
				connectionHolder.close(" heart beate idle ");
			}

		}

	}

	public int getIndex() {
		int currentIndex = 0;
		for (int i = 0; i < groupDataSource.getWriteDataSources().length; i++) {
			DataSource writeDataSource = groupDataSource.getWriteDataSources()[i];
			if (writeDataSource.getId().equals(getId())) {
				currentIndex = i;
				break;
			}
		}
		return currentIndex;
	}

	public boolean isSalveOrRead() {
		int currentIndex = getIndex();
		if (currentIndex != groupDataSource.currentActivedIndex || this.isReadNode) {
			return true;
		}
		return false;
	}

	public void heatBeatCheck(long timeout, long conHeartBeatPeriod) {
		// int ildeCloseCount = hostConfig.getMinCon() * 3;
		int maxConsInOneCheck = 5;
		LinkedList<IConnectionHolder> connectionHolderList = new LinkedList<IConnectionHolder>();

		long hearBeatTime = CurrentTimeMillis.currentTimeMillis() - conHeartBeatPeriod;
		long hearBeatTime2 = CurrentTimeMillis.currentTimeMillis() - 2 * conHeartBeatPeriod;
		for (ConnectionHolderQueue connectionHolderQueue : schemaConnectionHolderQueueMap.values()) {
			checkIfNeedHeartBeat(connectionHolderList, connectionHolderQueue, connectionHolderQueue.getConnectionHolderQueue(), hearBeatTime, hearBeatTime2);
			if (connectionHolderList.size() >= maxConsInOneCheck) {
				break;
			}
		}

		if (!connectionHolderList.isEmpty()) {
			for (IConnectionHolder connectionHolder : connectionHolderList) {
				conHeartBeatHanler.doHeartBeat(connectionHolder, groupDataSourceConfig.getHearbeatSQL());
			}
		}

		// check if there has timeouted heatbeat cons
		conHeartBeatHanler.abandTimeOuttedConns();
		int idleCount = getIdleCount();
		int activeCons = this.getActiveCount();
		int createCount = (groupDataSourceConfig.getMinSize() - idleCount) / 3;
		// create if idle too little
		if ((createCount > 0) && (idleCount + activeCons < maxSize) && (idleCount < groupDataSourceConfig.getMinSize())) {
			createByIdleLitte(idleCount, createCount);
		} else if (idleCount > groupDataSourceConfig.getMinSize()) {
			closeByIdleMany(idleCount - groupDataSourceConfig.getMinSize());
		} else {
			int activeCount = this.getActiveCount();
			if (activeCount > maxSize) {
				StringBuilder s = new StringBuilder();
				s.append(Alarms.DEFAULT).append("DATASOURCE EXCEED [name=").append(id).append(",active=");
				s.append(activeCount).append(",size=").append(maxSize).append(']');
				LOGGER.error(s.toString());
			}
		}
	}

	private void closeByIdleMany(int ildeCloseCount) {
		LOGGER.info("too many ilde cons ,close some for datasouce  " + id);
		List<IConnectionHolder> connectionHolderList = new ArrayList<IConnectionHolder>(ildeCloseCount);
		for (ConnectionHolderQueue connectionHolderQueue : schemaConnectionHolderQueueMap.values()) {
			connectionHolderList.addAll(connectionHolderQueue.getIdleConnectionHolderListToClose(ildeCloseCount));
			if (connectionHolderList.size() >= ildeCloseCount) {
				break;
			}
		}

		for (IConnectionHolder connectionHolder : connectionHolderList) {
			if (connectionHolder.isBorrowed()) {
				LOGGER.error("find idle con is using " + connectionHolder);
			}
			connectionHolder.close("too many idle con");
		}
	}

	private void createByIdleLitte(int idleCons, int createCount) {
		LOGGER.info("create connections ,because idle connection not enough ,cur is " + idleCons + ", minCon is " + groupDataSourceConfig.getMinSize() + " for " + id);
		NewConnectionRespHandler simpleHandler = new NewConnectionRespHandler();

		final String[] schemas = groupDataSource.getSchemas();
		for (int i = 0; i < createCount; i++) {
			if (this.getActiveCount() + this.getIdleCount() >= maxSize) {
				break;
			}
			try {
				// creat new connection
				this.createNewConnection(simpleHandler, null, schemas[i % schemas.length]);
			} catch (IOException e) {
				LOGGER.error("create connection err " + e);
			}

		}
	}

	public int getActiveCount() {
		return this.schemaConnectionHolderQueueMap.getActiveCountOfDataSource(this);
	}

	public void clearConnectionHolderList(String reason) {
		this.schemaConnectionHolderQueueMap.clearConnectionHolderList(reason, this);
	}

	public List<IConnectionHolder> shiftConnectionHolderList() {
		return this.schemaConnectionHolderQueueMap.shiftConnectionHolderList(this);
	}

	@SuppressWarnings("unused")
	private IConnectionHolder takeConnectionHolder(IConnectionHolder connectionHolder, final Handler handler, final Object attachment, String schema) {

		connectionHolder.setBorrowed(true);
		if (!connectionHolder.getSchema().equals(schema)) {
			connectionHolder.setSchema(schema);
		}
		ConnectionHolderQueue connectionHolderQueue = schemaConnectionHolderQueueMap.getConnectionHolderQueue(schema);
		connectionHolder.setAttachment(attachment);
		connectionHolder.setLastTime(System.currentTimeMillis()); // 每次取连接的时候，更新下lasttime，防止在前端连接检查的时候，关闭连接，导致sql执行失败
		handler.connectionAcquired(connectionHolder);
		return connectionHolder;
	}

	private void createNewConnection(final Handler handler, final Object attachment, final String schema) throws IOException {
		// aysn create connection
		Server.getInstance().getBusinessExecutor().execute(new Runnable() {
			public void run() {
				try {
					createNewConnection(new DelegateResponseHandler(handler) {

						@Override
						public void connectionAcquired(IConnectionHolder connectionHolder) {
							takeConnectionHolder(connectionHolder, handler, attachment, schema);
						}

						@Override
						public void connectionError(IConnectionHolder conn, Throwable e) {
							handler.connectionError(conn, e);
						}
					}, schema);
				} catch (IOException e) {
					handler.connectionError(null, e);
				}
			}
		});
	}

	public void getConnection(String schema, boolean autoCommit, final Handler responseHandler, final Object attachment) throws IOException {

		// 从当前连接map中拿取已建立好的后端连接
		IConnectionHolder connectionHolder = this.schemaConnectionHolderQueueMap.getConnectionHolder(schema);
		if (connectionHolder != null) {

			// 如果不为空，则绑定对应前端请求的handler
			takeConnectionHolder(connectionHolder, responseHandler, attachment, schema);
			return;

		} else {
			int activeCount = this.getActiveCount();// 当前最大活动连接
			if (activeCount + 1 > maxSize) {// 下一个连接大于最大连接数
				LOGGER.error("the max activeConnnections size can not be max than maxconnections");
				throw new IOException("the max activeConnnections size can not be max than maxconnections");
			} else { // create connection
				LOGGER.info("no ilde connection in pool,create new connection for " + this.id + " of schema " + schema);
				createNewConnection(responseHandler, attachment, schema);
			}
		}
	}

	private void returnConnectionHolder(IConnectionHolder connectionHolder) {

		connectionHolder.setAttachment(null);
		connectionHolder.setBorrowed(false);
		connectionHolder.setLastTime(CurrentTimeMillis.currentTimeMillis());
		ConnectionHolderQueue connectionHolderQueue = this.schemaConnectionHolderQueueMap.getConnectionHolderQueue(connectionHolder.getSchema());

		boolean ok = connectionHolderQueue.getConnectionHolderQueue().offer(connectionHolder);
		if (!ok) {
			LOGGER.error("can't return to pool ,so close con " + connectionHolder);
			connectionHolder.close("can't return to pool ");
		}
	}

	public void releaseChannel(IConnectionHolder connectionHolder) {
		returnConnectionHolder(connectionHolder);
	}

	public void connectionClosed(IConnectionHolder connectionHolder) {
		ConnectionHolderQueue connectionHolderQueue = this.schemaConnectionHolderQueueMap.getConnectionHolderQueue(connectionHolder.getSchema());
		if (connectionHolderQueue != null) {
			connectionHolderQueue.removeConnectionHolder(connectionHolder);
		}

	}

	public long getHeartbeatRecoveryTime() {
		return heartbeatRecoveryTime;
	}

	public void setHeartbeatRecoveryTime(long heartbeatRecoveryTime) {
		this.heartbeatRecoveryTime = heartbeatRecoveryTime;
	}

	public MysqlInstanceInfo getMysqlInstanceInfo() {
		return mysqlInstanceInfo;
	}

	public void createNewConnection(Handler handler, String schema) throws IOException {
		MysqlInstanceInfo mysqlInstanceInfo = getMysqlInstanceInfo();
		ConnectionHolder connectionHolder = new ConnectionHolder();
		connectionHolder.setHost(mysqlInstanceInfo.getIp());
		connectionHolder.setPort(mysqlInstanceInfo.getPort());
		connectionHolder.setDataSource(this);
		connectionHolder.setSchema(schema);

		SessionContextGroupShareResources sessionContextGroupShareResources = (SessionContextGroupShareResources) Server.getInstance().nextProcessor();
		connectionHolder.setSessionContextGroupShareResources(sessionContextGroupShareResources);
		connectionHolder.setId(1); // 复用mysql的Backend的ID，需要在process中存储

		sessionContextGroupShareResources.put(connectionHolder);
		try {
			Connection connection = getConnection();
			// c.setIdleTimeout(pool.getConfig().getIdleTimeout());
			connectionHolder.seConnection(connection);
			// notify handler
			handler.connectionAcquired(connectionHolder);
		} catch (Exception e) {
			handler.connectionError(connectionHolder, e);
		}
	}

	/**
	 * 当DRDS服务启动或者每次对数据源进行更改的时候进行测试。
	 */
	public boolean testConnection() throws IOException {
		// 假定多数的情况测试通过
		boolean isConnected = true;
		Connection connection = null;
		Statement statement = null;
		try {
			MysqlInstanceInfo $ = getMysqlInstanceInfo();
			connection = DriverManager.getConnection($.getUrl(), $.getUserName(), $.getPassword());
			statement = connection.createStatement();
			statement.execute("select 1");
		} catch (SQLException e) {
			e.printStackTrace();
			isConnected = false;
		} finally {
			if (statement != null) {
				try {
					statement.close();
				} catch (SQLException e) {
				}
			}

			if (connection != null) {
				try {
					connection.close();
				} catch (SQLException e) {
				}
			}
		}
		return isConnected;
	}

	Connection getConnection() throws SQLException {
		// $一般表示临时变量
		MysqlInstanceInfo $ = getMysqlInstanceInfo();
		Connection connection = DriverManager.getConnection(//
				$.getUrl(), //
				$.getUserName(),//
				$.getPassword());//
		return connection;
	}
}
